package com.kafkacode.producer;

import org.apache.kafka.clients.producer.ProducerRecord;
import org.apache.kafka.common.header.Headers;
import org.apache.kafka.common.header.internals.RecordHeaders;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.kafka.support.KafkaHeaders;
import org.springframework.messaging.Message;
import org.springframework.messaging.support.MessageBuilder;
import org.springframework.stereotype.Component;

import java.nio.charset.StandardCharsets;

@Component
public class EventProducer {
    @Autowired
    private KafkaTemplate<String,String> kafkaTemplate;

    public void sendEvent(){
        kafkaTemplate.send("topic-01","hello kafka");
    }
    public void sendEvent2(){
        Message<String> message = MessageBuilder.withPayload("hello kafka")
                .setHeader(KafkaHeaders.TOPIC,"test-topic")
                .build();
        kafkaTemplate.send(message);
    }

    public void sendEvent3(){
        Headers headers = new RecordHeaders();
        headers.add("phone","13240763332".getBytes(StandardCharsets.UTF_8));
        headers.add("orderId","1324".getBytes(StandardCharsets.UTF_8));
        ProducerRecord<String,String> record = new ProducerRecord<>(
                "topic-02",
                0,
                System.currentTimeMillis(),
                "k1",
                "hello kafka",
                headers);
        kafkaTemplate.send(record);
    }
}
